-module(rtps_spdp).

%%% @doc The Simple Participant Discovery Protocol (SPDP) process
%%% discovers the presence of other Participants on the network and
%%% their properties.

-behaviour(gen_statem).

-include_lib("kernel/include/logger.hrl").
-include("rtps.hrl").
-include("rtps_spdp.hrl").
-include("rtps_history_cache.hrl").

%% API
-export([start_link/2, setup/2, change/2, info/1]).

%% gen_statem callbacks
-export([init/1, callback_mode/0,
	 idle/3, 
	 announcing/3, 
	 handle_event/4, 
	 terminate/3, code_change/4]).

%%%===================================================================
%%% Types
%%%===================================================================
-export_type([discovered_participant_data/0, participant_proxy/0,
	      participant_builtin_topic_data/0]).
-type participant_proxy() :: #participant_proxy{}.
-type participant_builtin_topic_data() :: #participant_builtin_topic_data{}.
-type discovered_participant_data() :: #discovered_participant_data{}.

%%%===================================================================
%%% Defines and module records
%%%===================================================================

%% The detected remote participants
-record(participant, {guid_prefix :: rtps:guid_prefix(), last_seq_num :: rtps:sequence_number(),
		      discovered_participant_data :: #discovered_participant_data{}}).

%% internal state data
-record(data, {guid :: rtps:guid() | undefined, sup :: pid(), opts = [] :: [proplists:property()],
	       cache :: pid(), reg :: pid() | undefined, endpoints_sup :: pid() | undefined,
	       sedp :: pid() | undefined,
	       reader :: pid() | undefined, writer :: pid() | undefined,
	       resend_period :: rtps:duration(), lease_duration :: rtps:duration(),
	       participants :: [#participant{}]}).

%%%===================================================================
%%% API
%%%===================================================================

start_link(Sup, Opts) ->
    gen_statem:start_link(?MODULE, [Sup, Opts], []).

-spec setup(Pid, Participant_data) -> ok when
      Pid :: pid(),
      Participant_data :: rtps:discovered_participant_data().
%% @doc Set the `spdp_discovered_participant_data' by the participant
%% used by the SPDP.to announce the participant's existence.
setup(Pid, Participant_data) ->
    gen_statem:call(Pid, {setup, Participant_data}).

-spec change(Pid, Change) -> ok when
      Pid :: pid(),
      Change :: rtps_history_cache:cache_change().
%% @doc Process a cache change received from the reader which monitors
%% the remote participants. This can be a newly detected participant
%% or an already detected participant which re-announces its existence
%% on a regular basis with or without changed parameter values. An
%% incremented sequence number indicates that the values have
%% changed. But, it can also be a cache change indicating that the
%% remote participant was terminated.
-dialyzer([{no_match, change/2}]).
change(Pid, #cache_change{writer_guid = Writer_guid, sequence_number = Seq_num,
			  data_value = undefined, instance_handle = Serialized_key}) ->
    gen_statem:call(Pid, {remove, Writer_guid, Seq_num, Serialized_key});
change(Pid, #cache_change{writer_guid = Writer_guid, sequence_number = Seq_num,
			  data_value = Serialized_data}) ->
    gen_statem:call(Pid, {detect, Writer_guid, Seq_num, Serialized_data}).

info(Pid) ->
    gen_statem:call(Pid, info).

%%%===================================================================
%%% gen_statem callbacks
%%%===================================================================

%% @private
init([Sup, Opts]) ->
    process_flag(trap_exit, true),
    Opts1 = [{history_kind, keep_last}, {history_depth, 1}, {reliability_level, best_effort} | Opts],
    Cache = rtps_history_cache:new(Opts1),
    {ok, idle, #data{sup = Sup, opts = Opts, cache = Cache, participants = []}}.

%% @private
callback_mode() ->
    [state_functions, state_enter].

%% @private
idle(enter, _Old_state, #data{sup = Sup} = Data) ->
    Children = supervisor:which_children(Sup),
    {ok, Reg} = child(Children, registry),
    {ok, Endpoints_sup} = child(Children, endpoints),
    {ok, Sedp} = child(Children, sedp),
    {keep_state, Data#data{reg = Reg, endpoints_sup = Endpoints_sup, sedp = Sedp}};
idle({call, From}, {setup, #discovered_participant_data{participant_proxy = Participant_proxy,
							 lease_duration = Lease_duration} = Dpd},
     #data{cache = Cache, reg = Reg, sedp = Sedp, endpoints_sup = Endpoints_sup,
	   opts = Opts} = Data) ->
    Resend_period = proplists:get_value(resend_period, Opts, ?RESEND_PERIOD),
    Resend_period1 = min(Resend_period, Lease_duration),
    Opts1 = [{state_type, stateless}, {reliability_level, best_effort}, {history_cache, Cache},
	    {expects_inline_qos, false}, {topic_kind, with_key},
	    {multicast_locator_list, Participant_proxy#participant_proxy.metatraffic_multicast_locator_list},
	    {unicast_locator_list, Participant_proxy#participant_proxy.metatraffic_unicast_locator_list} | Opts],
    [begin
	 Spec = #{id => Entity_id,
		  start => {rtps_endpoint_sup, start_link, [Kind, Entity_id, Reg, Opts1]},
		  restart => permanent,
		  type => supervisor},
	 {ok, _} = rtps_ofo_sup:start_child(Endpoints_sup, Spec)
     end || {Kind, Entity_id} <- [{reader, 'SPDPbuiltinParticipantReader'},
				  {writer, 'SPDPbuiltinParticipantWriter'}]],
    Reader_pid = rtps_reg:whereis_name(Reg, 'SPDPbuiltinParticipantReader'),
    Writer_pid = rtps_reg:whereis_name(Reg, 'SPDPbuiltinParticipantWriter'),
    Serialized_data = enc_dpd(Dpd),
    %% Accordung to Fabrizio Ronci and Marco Listanti [Communications
    %% and Network, 2011, 3, 39-49]: The HistoryCache of the
    %% SPDPbuiltinParticipant Reader contains information on all
    %% active discovered Participants; the key used to identify each
    %% data-object corresponds to the Participant Globally Unique
    %% IDentifier (GUID). Lets use that as a reminder here.
    Guid = #guid{guid_prefix = Participant_proxy#participant_proxy.guid_prefix,
		   entity_id = 'participant'},
    {ok, Change} = rtps_writer:new_change(Writer_pid, alive, Serialized_data, Guid),
    %% {ok, Change} = rtps_writer:new_change(Writer_pid, alive, Serialized_data, undefined),
    rtps_writer:add_change(Writer_pid, Change),

    %% Install event handler in the history cache's event manager
    ok = rtps_history_cache:add_sup_handler(Cache, rtps_spdp_eh, []),

    Builtin_endpoints = Participant_proxy#participant_proxy.available_builtin_endpoints,
    rtps_sedp:setup(Sedp, Builtin_endpoints),
    
    {next_state, announcing,
     Data#data{guid = Guid, writer = Writer_pid, reader = Reader_pid,
	       resend_period = Resend_period1, lease_duration = Lease_duration},
     %% TODO: Do document that we take the smallest of the
     %% Resend_period and Lease_duration as the resend period, or
     %% remove this. This is maybe to much automagic behaviour.
     [{{timeout, resend_period}, Resend_period1, undefined},
      {reply, From, ok}]}.

announcing({call, From}, {detect, #guid{guid_prefix = Guid_prefix}, Seq_num, Serialized_data},
	   #data{sedp = Sedp, participants = Participants} = Data) ->
    Dpd = dec_dpd(Serialized_data),
    Lease_duration = Dpd#discovered_participant_data.lease_duration,
    case lists:keytake(Guid_prefix, #participant.guid_prefix, Participants) of
	{value, #participant{last_seq_num = Last_seq_num} = Participant, Participants1}
	  when Seq_num > Last_seq_num ->
	    ok = rtps_sedp:update(Sedp, Dpd#discovered_participant_data.participant_proxy),
	    Participant1 = Participant#participant{last_seq_num = Seq_num,
						   discovered_participant_data = Dpd},
	    {keep_state, Data#data{participants = [Participant1 | Participants1]},
	     [{{timeout, Guid_prefix}, Lease_duration, lease_duration}, {reply, From, ok}]};
	{value, #participant{}, _} ->
	    {keep_state_and_data,
	     [{{timeout, Guid_prefix}, Lease_duration, lease_duration}, {reply, From, ok}]};
	false ->
	    Dpd = dec_dpd(Serialized_data),
	    ok = rtps_sedp:add(Sedp, Dpd#discovered_participant_data.participant_proxy),
	    Participant = #participant{guid_prefix = Guid_prefix, last_seq_num = Seq_num,
				       discovered_participant_data = Dpd},
	    %% Also act on detecting a new participant by re-announcing our own existence
	    {keep_state, Data#data{participants = [Participant | Participants]},
	     [{{timeout, Guid_prefix}, Lease_duration, lease_duration},
	      {{timeout, resend_period}, 0, undefined}, {reply, From, ok}]}
    end;
announcing({call, From}, {remove, _Writer_guid, _Seq_num, Serialized_key},
	  #data{sedp = Sedp, participants = Participants} = Data) ->
    #guid{guid_prefix = Guid_prefix} = rtps_psm:dec_guid(Serialized_key),
    case lists:keytake(Guid_prefix, #participant.guid_prefix, Participants) of
	{value, #participant{guid_prefix = Guid_prefix}, Participants1} ->
	    rtps_sedp:remove(Sedp, Guid_prefix),
	    {keep_state, Data#data{participants = Participants1},
	     [{{timeout, Guid_prefix}, infinity, lease_duration}, {reply, From, ok}]};
	false ->
	    keep_state_and_data
    end;
announcing({timeout, resend_period}, _Content,
	  #data{writer = Writer_pid, resend_period = Resend_period}) ->
    rtps_writer:unsent_changes_reset(Writer_pid),
    {keep_state_and_data, [{{timeout, resend_period}, Resend_period, undefined}]};
announcing({timeout, Guid_prefix}, lease_duration,
	  #data{sedp = Sedp, cache = Cache, participants = Participants} = Data) ->
    case lists:keytake(Guid_prefix, #participant.guid_prefix, Participants) of
	{value, #participant{guid_prefix = Guid_prefix}, Participants1} ->
	    rtps_sedp:remove(Sedp, Guid_prefix),
	    Guid = #guid{guid_prefix = Guid_prefix, entity_id = 'SPDPbuiltinParticipantWriter'},
	    %% TODO: Do not remove but perform unregister?!
	    rtps_history_cache:remove_changes(Cache, Guid),
	    {keep_state, Data#data{participants = Participants1}};
	false ->
	    keep_state_and_data
    end;
announcing(Event_type, Event_content, Data) ->
    handle_event(Event_type, Event_content, announcing, Data).

%% @private
handle_event({call, From}, info, _State, 
	     #data{participants = Participants}) ->
    Reply = [Guid_prefix || #participant{guid_prefix = Guid_prefix} <- Participants],
    {keep_state_and_data, [{reply, From, Reply}]};

%% TODO: remove catch-all clauses
handle_event({call, From} = _Event_type, _Event_content, _State_name, _Data) ->
    Reply = ok,
    {keep_state_and_data, [{reply, From, Reply}]};
handle_event(_Event_type,  _Event_content, _State_name, _Data) ->
    keep_state_and_data.

%% @private
terminate(_Reason, _State_name, #data{guid = Guid, writer = Writer_pid}) 
  when Guid =/= undefined ->
    Handle = rtps_psm:enc_guid(Guid),
    {ok, Change} = rtps_writer:new_change(Writer_pid, not_alive_disposed, undefined, Handle),
    rtps_writer:add_change(Writer_pid, Change);
terminate(_Reason, _State_name, _Data) ->
    ok.

%% @private
code_change(_Old_vsn, State_name, Data, _Extra) ->
    {ok, State_name, Data}.

%%%===================================================================
%%% Internal functions
%%%===================================================================

%% @private
child(Children, Id) ->
    case lists:keyfind(Id, 1, Children) of
	{_Id, Child, _Type, _Modules} when is_pid(Child) ->
	    {ok, Child};
	{_Id, Child, _Type, _Modules} when Child =:= restarting ->
	    {error, Child};
	_ ->
	    {error, notfound}
    end.

%% @private
enc_dpd(#discovered_participant_data{participant_proxy = Participant_proxy,
				     participant_builtin_topic_data = Qos,
				     lease_duration = Lease_duration}) ->
    #participant_proxy {protocol_version = Protocol_version,
			guid_prefix = Guid_prefix,
			vendor_id = Vendor_id,
			expects_inline_qos = Expects_inline_qos,
			metatraffic_unicast_locator_list = Meta_u_locs,
			metatraffic_multicast_locator_list = Meta_m_locs,
			default_unicast_locator_list = U_locs,
			default_multicast_locator_list = M_locs,
			available_builtin_endpoints = Available_builtin_endpoints,
			manual_liveliness_count = Manual_liveliness_count} = Participant_proxy,
    Guid = #guid{guid_prefix = Guid_prefix, entity_id = participant},
    Parameter_list = 
	[#parameter{id = protocol_version, value = Protocol_version},
	 #parameter{id = participant_guid, value = Guid},
	 #parameter{id = vendor_id, value = Vendor_id},
	 #parameter{id = expects_inline_qos, value = Expects_inline_qos}] ++
	[#parameter{id = metatraffic_unicast_locator, value = Loc} || Loc <- Meta_u_locs] ++
	[#parameter{id = metatraffic_multicast_locator, value = Loc} || Loc <- Meta_m_locs] ++
	[#parameter{id = default_unicast_locator, value = Loc} || Loc <- U_locs] ++
	[#parameter{id = default_multicast_locator, value = Loc} || Loc <- M_locs] ++
	[#parameter{id = builtin_endpoint_set, value = Available_builtin_endpoints},
	 #parameter{id = participant_lease_duration, value = rtps:convtime_ms(Lease_duration)},
	 #parameter{id = participant_manual_liveliness_count, value = Manual_liveliness_count}],
    rtps_psm:enc_data(Parameter_list ++ Qos).
    
-dialyzer([{no_match, dec_dpd/1}]).
%% @private
dec_dpd(Bin) ->
    {Parameter_list, _Bin} = rtps_psm:dec_data(Bin),
    Fun = % Process participant proxy parameters
	fun(#parameter{id = protocol_version, value = Protocol_version},
	    {Participant_proxy, Qos, Lease_duration}) ->
		{Participant_proxy#participant_proxy{protocol_version = Protocol_version},
		 Qos, Lease_duration};
	   (#parameter{id = participant_guid, value = #guid{guid_prefix = Guid_prefix}},
	    {Participant_proxy, Qos, Lease_duration}) ->
		{Participant_proxy#participant_proxy{guid_prefix = Guid_prefix},
		 Qos, Lease_duration};
	   (#parameter{id = vendor_id, value = Vendor_id},
	    {Participant_proxy, Qos, Lease_duration}) ->
		{Participant_proxy#participant_proxy{vendor_id = Vendor_id},
		 Qos, Lease_duration};
	   (#parameter{id = expects_inline_qos, value = Expects_inline_qos},
	    {Participant_proxy, Qos, Lease_duration}) ->
		{Participant_proxy#participant_proxy{expects_inline_qos = Expects_inline_qos},
		 Qos, Lease_duration};
	   (#parameter{id = metatraffic_unicast_locator, value = Loc},
	    {#participant_proxy{metatraffic_unicast_locator_list = Locs} = Participant_proxy,
	     Qos, Lease_duration})
	      when is_list(Locs) ->
		{Participant_proxy#participant_proxy{metatraffic_unicast_locator_list = [Loc | Locs]},
		 Qos, Lease_duration};
	   (#parameter{id = metatraffic_unicast_locator, value = Loc},
	    {Participant_proxy, Qos, Lease_duration}) ->
		{Participant_proxy#participant_proxy{metatraffic_unicast_locator_list = [Loc]},
		 Qos, Lease_duration};
	   (#parameter{id = metatraffic_multicast_locator, value = Loc},
	    {#participant_proxy{metatraffic_multicast_locator_list = Locs} = Participant_proxy,
	     Qos, Lease_duration})
	      when is_list(Locs) ->
		{Participant_proxy#participant_proxy{metatraffic_multicast_locator_list = [Loc | Locs]}, Qos, Lease_duration};
	   (#parameter{id = metatraffic_multicast_locator, value = Loc},
	    {Participant_proxy, Qos, Lease_duration}) ->
		{Participant_proxy#participant_proxy{metatraffic_multicast_locator_list = [Loc]},
		 Qos, Lease_duration};
	   (#parameter{id = default_unicast_locator, value = Loc},
	    {#participant_proxy{default_unicast_locator_list = Locs} = Participant_proxy,
	     Qos, Lease_duration})
	      when is_list(Locs) ->
		{Participant_proxy#participant_proxy{default_unicast_locator_list = [Loc | Locs]},
		 Qos, Lease_duration};
	   (#parameter{id = default_unicast_locator, value = Loc},
	    {Participant_proxy, Qos, Lease_duration}) ->
		{Participant_proxy#participant_proxy{default_unicast_locator_list = [Loc]},
		 Qos, Lease_duration};
	   (#parameter{id = default_multicast_locator, value = Loc},
	    {#participant_proxy{default_multicast_locator_list = Locs} = Participant_proxy,
	     Qos, Lease_duration})
	      when is_list(Locs) ->
		{Participant_proxy#participant_proxy{default_multicast_locator_list = [Loc | Locs]},
		 Qos, Lease_duration};
	   (#parameter{id = default_multicast_locator, value = Loc},
	    {Participant_proxy, Qos, Lease_duration}) ->
		{Participant_proxy#participant_proxy{default_multicast_locator_list = [Loc]},
		 Qos, Lease_duration};
	   (#parameter{id = builtin_endpoint_set, value = Available_builtin_endpoints},
	    {Participant_proxy, Qos, Lease_duration}) ->
		{Participant_proxy#participant_proxy{available_builtin_endpoints = Available_builtin_endpoints},
		 Qos, Lease_duration};
	   (#parameter{id = participant_manual_liveliness_count, value = Manual_liveliness_count},
	    {Participant_proxy, Qos, Lease_duration}) ->
		{Participant_proxy#participant_proxy{manual_liveliness_count = Manual_liveliness_count},
		 Qos, Lease_duration};
	   %% Process lease duration
	   (#parameter{id = participant_lease_duration, value = Time},
	    {Participant_proxy, Qos, _Lease_duration}) ->
		Lease_duration = rtps:convtime_ms(Time),
		{Participant_proxy, Qos, Lease_duration};
	   %% Process ParticipantBuiltinTopicData
	   (Parameter, {Particiapnt_proxy, Qos, Lease_duration}) ->
		{Particiapnt_proxy, [Parameter | Qos], Lease_duration}
	end,
    %% The next line gives a dialyzer error because the
    %% #participant_proxy{} record is created without the proper
    %% values for the fields. The fields are filled in while
    %% processing the parameter list. Ignore this plt warning
    {Participant_proxy, Qos, Lease_duration} = lists:foldl(Fun, {#participant_proxy{}, [], undefined}, Parameter_list),
    #discovered_participant_data{participant_proxy = Participant_proxy,
				 participant_builtin_topic_data = Qos,
				 lease_duration = Lease_duration}.

